package org.shuzhou.f_watermark;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.Date;

/**
 * 使用事件时间，watermark案例
 * 数据格式：2023-12-14 08:30:12 - INFO 用户登录成功，用户名：JohnDoe
 */
public class LogProcessingJob {

    public static void main(String[] args) throws Exception {
        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 获取日志数据流
        DataStream<String> logStream = env.fromElements(
                "2023-12-14 08:30:12 - INFO 用户登录成功，用户名：JohnDoe",
                "2023-12-14 09:15:29 - INFO 用户查看了产品列表页面",
                "2023-12-14 10:02:45 - INFO 用户添加了商品到购物车，商品ID：12345",
                "2023-12-14 11:20:01 - INFO 用户提交了订单，订单号：67890",
                "2023-12-14 13:45:18 - INFO 用户收到订单确认邮件",
                "2023-12-14 14:30:55 - INFO 用户浏览了帮助中心页面",
                "2023-12-14 15:10:27 - INFO 用户退出登录");

        // 处理日志数据流
        DataStream<String> processedLogStream = logStream
                .assignTimestampsAndWatermarks(getLogWatermarkStrategy())
                .map(log -> "Processed log: " + log);

        // 输出结果到控制台
        processedLogStream.print();

        // 执行任务
        env.execute("Log Processing Job");
    }

    /**
     * 获取日志数据的 WatermarkStrategy，用于生成水印和指定事件时间。
     */
    private static WatermarkStrategy<String> getLogWatermarkStrategy() {
        return WatermarkStrategy
                // 最大乱序程度为 3 秒的水印策略
                .<String>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                // 提供自定义的时间戳分配器
                .withTimestampAssigner((log, timestamp) -> {
                    // 提取事件时间
                    SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    Date eventTime;
                    try {
                        eventTime = format.parse(log.substring(0, 19));
                    } catch (ParseException e) {
                        throw new RuntimeException("Unable to parse event time from log: " + log, e);
                    }
                    return eventTime.getTime();
                });
    }
}